package jupitermouse.site.example.scala.batch

import jupitermouse.site.example.java.entity.People
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration

object DataSetDataSourceApp {

  def main(args: Array[String]): Unit = {

    val env = ExecutionEnvironment.getExecutionEnvironment
//    fromCollection(env)

//    fromTextFile(env)

//    fromcsvFile(env)

//    fromreadRecursiveFiles(env) 递归文件夹的内容

    fromCompressionFiles(env) //读取压缩文件
  }

  def fromCollection(env:ExecutionEnvironment): Unit = {
    import org.apache.flink.api.scala._
    val data = 1 to 10
    env.fromCollection(data).print()
  }

  def fromTextFile(env:ExecutionEnvironment):Unit = {
//    val inputPath = "file:///E:\\workroom\\learn\\flink\\demo_flink\\doc\\word.txt"
    val inputPath = "file:///E:\\workroom\\learn\\flink\\demo_flink\\doc"
    env.readTextFile(inputPath).print()
  }

  def fromcsvFile(env:ExecutionEnvironment):Unit = {
    import org.apache.flink.api.scala._
    val inputPath = "file:///E:\\workroom\\learn\\flink\\demo_flink\\docadd\\csv\\people.csv";
//    env.readCsvFile[(String,Int,String)](inputPath, ignoreFirstLine=true,fieldDelimiter=";").print()

//    env.readCsvFile[(String,Int,String)](inputPath, ignoreFirstLine=true,fieldDelimiter=";").print()

    case class ScalaPeople(name:String, age:Int, job:String)

    //要使用pojo
    env.readCsvFile[People](inputPath, ignoreFirstLine=true,fieldDelimiter=";",
      pojoFields = Array("name","age","job")).print()
  }

  def fromreadRecursiveFiles(env:ExecutionEnvironment):Unit = {
    val inputFile = "file:///E:\\workroom\\learn\\flink\\demo_flink\\doc"
    env.readTextFile(inputFile).print()
    println("-------split----------")
    val parameters = new Configuration
    parameters.setBoolean("recursive.file.enumeration", true)
    env.readTextFile(inputFile).withParameters(parameters).print()
  }

  def fromCompressionFiles(env:ExecutionEnvironment):Unit = {
    val fileInput = "file:///E:\\workroom\\learn\\flink\\demo_flink\\doc\\people.zip"
    env.readTextFile(fileInput).print()

  }

}
